//! TCP服务器模块
//! 
//! 这个模块实现了TCP服务器功能，包括：
//! - TCP连接管理
//! - 自定义协议处理
//! - 消息广播
//! - 连接状态监控

use tokio::net::{TcpListener, TcpStream};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::RwLock;
use std::collections::HashMap;
use std::time::{Duration, Instant};
use uuid::Uuid;
use bytes::{Bytes, BytesMut, Buf, BufMut};

use crate::{
    config::{WebSocketConfig, TcpConfig},
    errors::AppError,
    AppState,
};

/// TCP连接信息
#[derive(Debug, Clone)]
pub struct TcpConnection {
    /// 连接ID
    pub id: Uuid,
    /// 客户端地址
    pub addr: SocketAddr,
    /// 连接时间
    pub connected_at: Instant,
    /// 最后活动时间
    pub last_activity: Instant,
    /// 是否活跃
    pub is_alive: bool,
}

/// TCP服务器
pub struct TcpServer {
    /// 连接管理器
    connections: Arc<RwLock<HashMap<Uuid, TcpConnection>>>,
    /// 服务器配置
    config: TcpConfig,
}

impl TcpServer {
    /// 创建新的WebSocket TCP服务器
    pub async fn new_websocket(config: &WebSocketConfig) -> Result<Self, std::io::Error> {
        Ok(Self {
            connections: Arc::new(RwLock::new(HashMap::new())),
            config: TcpConfig {
                port: config.port,
                max_connections: config.max_connections,
                buffer_size: 4096,
            },
        })
    }

    /// 创建新的TCP服务器
    pub async fn new_tcp(config: &TcpConfig) -> Result<Self, std::io::Error> {
        Ok(Self {
            connections: Arc::new(RwLock::new(HashMap::new())),
            config: config.clone(),
        })
    }

    /// 运行WebSocket TCP服务器
    pub async fn run_websocket(&self, state: AppState) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let addr = SocketAddr::from(([0, 0, 0, 0], self.config.port));
        let listener = TcpListener::bind(addr).await?;
        
        tracing::info!("🔌 WebSocket TCP服务器启动在 {}", addr);

        loop {
            match listener.accept().await {
                Ok((stream, addr)) => {
                    tracing::info!("📡 新WebSocket TCP连接: {}", addr);

                    // 检查连接数限制
                    {
                        let connections = self.connections.read().await;
                        if connections.len() >= self.config.max_connections {
                            tracing::warn!("⚠️ WebSocket TCP连接数已达上限，拒绝连接: {}", addr);
                            continue;
                        }
                    }

                    // 创建连接ID
                    let connection_id = Uuid::new_v4();
                    
                    // 添加到连接管理器
                    {
                        let mut connections = self.connections.write().await;
                        connections.insert(connection_id, TcpConnection {
                            id: connection_id,
                            addr,
                            connected_at: Instant::now(),
                            last_activity: Instant::now(),
                            is_alive: true,
                        });
                    }

                    // 异步处理连接
                    let connections = Arc::clone(&self.connections);
                    let state = state.clone();

                    tokio::spawn(async move {
                        if let Err(e) = handle_websocket_tcp_connection(stream, addr, connection_id, connections, state).await {
                            tracing::error!("❌ WebSocket TCP连接处理错误: {}", e);
                        }
                    });
                }
                Err(e) => {
                    tracing::error!("❌ 接受WebSocket TCP连接失败: {}", e);
                }
            }
        }
    }

    /// 运行TCP服务器
    pub async fn run_tcp(&self, state: AppState) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let addr = SocketAddr::from(([0, 0, 0, 0], self.config.port));
        let listener = TcpListener::bind(addr).await?;
        
        tracing::info!("🌐 TCP服务器启动在 {}", addr);

        loop {
            match listener.accept().await {
                Ok((stream, addr)) => {
                    tracing::info!("📡 新TCP连接: {}", addr);

                    // 检查连接数限制
                    {
                        let connections = self.connections.read().await;
                        if connections.len() >= self.config.max_connections {
                            tracing::warn!("⚠️ TCP连接数已达上限，拒绝连接: {}", addr);
                            continue;
                        }
                    }

                    // 创建连接ID
                    let connection_id = Uuid::new_v4();
                    
                    // 添加到连接管理器
                    {
                        let mut connections = self.connections.write().await;
                        connections.insert(connection_id, TcpConnection {
                            id: connection_id,
                            addr,
                            connected_at: Instant::now(),
                            last_activity: Instant::now(),
                            is_alive: true,
                        });
                    }

                    // 异步处理连接
                    let connections = Arc::clone(&self.connections);
                    let state = state.clone();

                    tokio::spawn(async move {
                        if let Err(e) = handle_tcp_connection(stream, addr, connection_id, connections, state).await {
                            tracing::error!("❌ TCP连接处理错误: {}", e);
                        }
                    });
                }
                Err(e) => {
                    tracing::error!("❌ 接受TCP连接失败: {}", e);
                }
            }
        }
    }

    /// 获取连接统计信息
    pub async fn get_connection_stats(&self) -> (usize, usize) {
        let connections = self.connections.read().await;
        let total = connections.len();
        let active = connections.values().filter(|conn| conn.is_alive).count();
        (total, active)
    }

    /// 清理死连接
    pub async fn cleanup_dead_connections(&self) -> usize {
        let mut connections = self.connections.write().await;
        let timeout_duration = Duration::from_secs(60);
        let now = Instant::now();

        let mut dead_connections = Vec::new();

        for (id, connection) in connections.iter_mut() {
            if now.duration_since(connection.last_activity) > timeout_duration {
                connection.is_alive = false;
                dead_connections.push(*id);
            }
        }

        let count = dead_connections.len();
        for id in dead_connections {
            connections.remove(&id);
        }

        count
    }
}

/// 处理WebSocket TCP连接
async fn handle_websocket_tcp_connection(
    stream: TcpStream,
    addr: SocketAddr,
    connection_id: Uuid,
    connections: Arc<RwLock<HashMap<Uuid, TcpConnection>>>,
    state: AppState,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let (mut reader, mut writer) = stream.into_split();
    
    tracing::info!("🔗 开始处理WebSocket TCP连接: {} ({})", addr, connection_id);

    let mut buffer = BytesMut::with_capacity(4096);

    loop {
        // 读取数据
        let bytes_read = tokio::time::timeout(
            Duration::from_secs(30),
            tokio::io::AsyncReadExt::read(&mut reader, &mut buffer)
        ).await??;

        if bytes_read == 0 {
            tracing::info!("🔚 WebSocket TCP连接关闭: {}", addr);
            break;
        }

        // 更新最后活动时间
        {
            let mut connections = connections.write().await;
            if let Some(connection) = connections.get_mut(&connection_id) {
                connection.last_activity = Instant::now();
            }
        }

        // 处理接收到的数据
        let data = &buffer[..bytes_read];
        tracing::debug!("📥 收到WebSocket TCP数据: {} bytes from {}", bytes_read, addr);

        // 简单的echo服务器
        let response = format!("WebSocket TCP Echo: {}\n", String::from_utf8_lossy(data));
        
        // 发送响应
        tokio::time::timeout(
            Duration::from_secs(10),
            tokio::io::AsyncWriteExt::write_all(&mut writer, response.as_bytes())
        ).await??;

        tracing::debug!("📤 发送WebSocket TCP响应: {} bytes to {}", response.len(), addr);

        // 清空缓冲区
        buffer.clear();
    }

    // 清理连接
    {
        let mut connections = connections.write().await;
        connections.remove(&connection_id);
    }

    tracing::info!("🧹 WebSocket TCP连接已清理: {}", connection_id);
    Ok(())
}

/// 处理TCP连接
async fn handle_tcp_connection(
    stream: TcpStream,
    addr: SocketAddr,
    connection_id: Uuid,
    connections: Arc<RwLock<HashMap<Uuid, TcpConnection>>>,
    state: AppState,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let (mut reader, mut writer) = stream.into_split();
    
    tracing::info!("🔗 开始处理TCP连接: {} ({})", addr, connection_id);

    let mut buffer = BytesMut::with_capacity(4096);

    loop {
        // 读取数据
        let bytes_read = tokio::time::timeout(
            Duration::from_secs(30),
            tokio::io::AsyncReadExt::read(&mut reader, &mut buffer)
        ).await??;

        if bytes_read == 0 {
            tracing::info!("🔚 TCP连接关闭: {}", addr);
            break;
        }

        // 更新最后活动时间
        {
            let mut connections = connections.write().await;
            if let Some(connection) = connections.get_mut(&connection_id) {
                connection.last_activity = Instant::now();
            }
        }

        // 处理接收到的数据
        let data = &buffer[..bytes_read];
        tracing::debug!("📥 收到TCP数据: {} bytes from {}", bytes_read, addr);

        // 处理自定义协议
        let response = process_tcp_message(data, &state).await;

        // 发送响应
        if let Some(response_data) = response {
            tokio::time::timeout(
                Duration::from_secs(10),
                tokio::io::AsyncWriteExt::write_all(&mut writer, &response_data)
            ).await??;

            tracing::debug!("📤 发送TCP响应: {} bytes to {}", response_data.len(), addr);
        }

        // 清空缓冲区
        buffer.clear();
    }

    // 清理连接
    {
        let mut connections = connections.write().await;
        connections.remove(&connection_id);
    }

    tracing::info!("🧹 TCP连接已清理: {}", connection_id);
    Ok(())
}

/// 处理TCP消息
async fn process_tcp_message(data: &[u8], state: &AppState) -> Option<Vec<u8>> {
    let message = String::from_utf8_lossy(data);
    let message = message.trim();

    match message {
        "PING" => {
            Some(b"PONG\n".to_vec())
        }
        "STATS" => {
            let total_connections = state.websocket.get_online_users_count().await;
            let active_connections = 0; // TCP连接统计暂时设为0
            let (db_total, db_idle) = state.db.get_pool_info();
            let (cache_total, cache_idle) = state.cache.get_pool_info();
            
            let stats = format!(
                "STATS\nWebSocket Connections: {}\nDatabase Pool: {}/{}\nCache Pool: {}/{}\n",
                total_connections, db_total, db_idle, cache_total, cache_idle
            );
            Some(stats.into_bytes())
        }
        "TIME" => {
            let time = chrono::Utc::now().to_rfc3339();
            Some(format!("TIME {}\n", time).into_bytes())
        }
        _ => {
            // 默认echo
            Some(format!("ECHO: {}\n", message).into_bytes())
        }
    }
}

/// TCP消息类型枚举
#[derive(Debug, Clone)]
pub enum TcpMessageType {
    Ping,
    Pong,
    Stats,
    Time,
    Echo,
    Error,
}

/// TCP消息结构
#[derive(Debug, Clone)]
pub struct TcpMessage {
    pub message_type: TcpMessageType,
    pub content: String,
    pub timestamp: chrono::DateTime<chrono::Utc>,
}

impl TcpMessage {
    /// 创建新的TCP消息
    pub fn new(message_type: TcpMessageType, content: String) -> Self {
        Self {
            message_type,
            content,
            timestamp: chrono::Utc::now(),
        }
    }

    /// 序列化为字节
    pub fn to_bytes(&self) -> Vec<u8> {
        let json = serde_json::json!({
            "type": format!("{:?}", self.message_type),
            "content": self.content,
            "timestamp": self.timestamp.to_rfc3339()
        });
        format!("{}\n", json).into_bytes()
    }

    /// 从字节反序列化
    pub fn from_bytes(data: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
        let json_str = String::from_utf8_lossy(data);
        let json: serde_json::Value = serde_json::from_str(&json_str)?;
        
        let message_type = match json["type"].as_str().unwrap_or("") {
            "Ping" => TcpMessageType::Ping,
            "Pong" => TcpMessageType::Pong,
            "Stats" => TcpMessageType::Stats,
            "Time" => TcpMessageType::Time,
            "Echo" => TcpMessageType::Echo,
            _ => TcpMessageType::Error,
        };

        Ok(Self {
            message_type,
            content: json["content"].as_str().unwrap_or("").to_string(),
            timestamp: chrono::DateTime::parse_from_rfc3339(json["timestamp"].as_str().unwrap_or(""))
                .unwrap_or_default()
                .with_timezone(&chrono::Utc),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_tcp_message_serialization() {
        let message = TcpMessage::new(TcpMessageType::Ping, "test".to_string());
        let bytes = message.to_bytes();
        let deserialized = TcpMessage::from_bytes(&bytes).unwrap();

        assert_eq!(message.content, deserialized.content);
        assert!(matches!(deserialized.message_type, TcpMessageType::Ping));
    }

    #[tokio::test]
    async fn test_tcp_server_creation() {
        let config = TcpConfig::default();
        let server = TcpServer::new_tcp(&config).await.unwrap();
        
        let (total, active) = server.get_connection_stats().await;
        assert_eq!(total, 0);
        assert_eq!(active, 0);
    }

    #[tokio::test]
    async fn test_tcp_message_processing() {
        // 这里需要模拟AppState，为了简化测试，我们跳过具体实现
        // 在实际测试中，应该创建模拟的AppState
        assert!(true); // 占位测试
    }
}
